#ifndef MQSERVER_MQ_CONSUMER_HPP_
#define MQSERVER_MQ_CONSUMER_HPP_
#include "../mqcomm/proto/mq_msg.pb.h"
#include "../mqcomm/mq_logger.hpp"
#include <functional>
#include <memory>
#include <mutex>
#include <sstream>
#include <string>
#include <unordered_map>
#include <vector>

namespace mqserver 
{

using ConsumerCallback = std::function<void(const std::string,  // 消费者标识
                            const proto::message::BasicProperties*,     // 消息基本属性
                            const std::string)>;                // 消息正文

class Consumer;
class QueueConsumer;
class ConsumerManager;
using ConsumerPtr = std::shared_ptr<Consumer>;
using QueueConsumerPtr = std::shared_ptr<QueueConsumer>;
using ConsumerManagerPtr = std::shared_ptr<ConsumerManager>;

// 消费者数据结构
class Consumer  
{
public:
    Consumer(const std::string& tag, const std::string& qname, 
             bool auto_ack, const ConsumerCallback& callback)
        : tag_(tag), qname_(qname), auto_ack_(auto_ack), callback_(std::move(callback)) {}

    std::string tag_;     // 消费者标识
    std::string qname_;   // 消费者订阅的队列名称
    bool auto_ack_;       // 自动确认标志
    ConsumerCallback callback_; 
};

// 消费者管理类
class QueueConsumer
{
public:
    QueueConsumer(const std::string& qname)
        : qname_(qname), rr_seq(0) {}

    ConsumerPtr create(const std::string& tag, const std::string& qname,    // 新增消费者
                bool auto_ack, const ConsumerCallback& callback)
    {
        std::lock_guard<std::mutex> lock(mtx_);
        // 1. 重复判断(存在就返回)
        if(findConsumerByTag(tag) != consumers_.end()) {
            return ConsumerPtr(); 
        }
        // 2. 构造对象
        consumers_.emplace_back(std::make_shared<Consumer>(tag, qname, auto_ack, callback));
        return consumers_.back();
    }
    bool remove(const std::string& tag)    // 删除消费者(取消订阅/信道关闭/连接关闭时删除)
    {
        std::lock_guard<std::mutex> lock(mtx_);
        ConsumerIterator it = findConsumerByTag(tag);
        if(it != consumers_.end()) {
            consumers_.erase(it);
            return true;
        }
        return false;
    }
    ConsumerPtr getConsumer()   // 获取消费者：从队列所有的消费者中按序取出一个消费者进行消息的推送
    {
        std::lock_guard<std::mutex> lock(mtx_);
        // 1. 预处理
        if(consumers_.size() == 0) return ConsumerPtr();
        // 2. RR轮转
        int idx = rr_seq % consumers_.size();
        ++rr_seq;
        // 3. 返回
        return consumers_[idx];
    }
    bool emptyQueue()   // 判断队列消费者是否为空
    {
        std::lock_guard<std::mutex> lock(mtx_);
        return consumers_.size() == 0;
    }
    bool exists(const std::string& tag)  // 判断指定消费者是否存在
    {
        std::lock_guard<std::mutex> lock(mtx_);
        if(findConsumerByTag(tag) != consumers_.end()) return true;
        return false;
    }
    void clear()
    {
        std::lock_guard<std::mutex> lock(mtx_);
        consumers_.clear();
        rr_seq = 0;
    }
private:
    using ConsumerIterator = std::vector<ConsumerPtr>::iterator;
    ConsumerIterator findConsumerByTag(const std::string& tag)
    {
        auto it = consumers_.begin();
        for (; it != consumers_.end(); ++it) {
            if((*it)->tag_ == tag) break;
        }
        return it;
    }
private:
    std::mutex mtx_;
    std::string qname_;   // 队列名称
    uint64_t rr_seq;      // 轮转序号
    std::vector<ConsumerPtr> consumers_;    
};

// 以队列为单元管理消费者
class ConsumerManager 
{
public:
    ConsumerManager() = default;
    void initQueueConsumer(const std::string& qname)    // 初始化队列的消费者信息结构
    {
        std::lock_guard<std::mutex> lock(mtx_);
        // 1. 去重
        if(existsQueue(qname)) return;
        // 2. 构造对象
        que_consumers_.emplace(qname, std::make_shared<QueueConsumer>(qname));
    }
    void destoryQueueConsumer(const std::string& qname)  // 删除队列的消费者信息结构
    {
        std::lock_guard<std::mutex> lock(mtx_);
        que_consumers_.erase(qname);
    }
    ConsumerPtr create(const std::string& tag, const std::string& qname,    // 新增消费者
                bool auto_ack, const ConsumerCallback& callback)
    {
        QueueConsumerPtr que_consumer = getQueueConsumerPtr(qname);
        if(que_consumer == nullptr) {
            LOG_ERROR << "insert consumer to the \"" << qname << "\" fail! No queue_consumer handle found!\n";
            return ConsumerPtr();
        }
        return que_consumer->create(tag, qname, auto_ack, callback);  
    }
    bool remove(const std::string& qname, const std::string& tag)   // 从指定队列移除消费者
    {
        QueueConsumerPtr que_consumer = getQueueConsumerPtr(qname);
        if(que_consumer == nullptr) {
            LOG_ERROR << "remove consumer from the \"" << qname << "\" fail! No queue_consumer handle found!\n";
            return false;
        }
        return que_consumer->remove(tag);
    }
    ConsumerPtr getConsumer(const std::string& qname)   // 轮询获取消费者, 起到负载均衡的作用
    {
        QueueConsumerPtr que_consumer = getQueueConsumerPtr(qname);
        if(que_consumer == nullptr) {
            LOG_ERROR << "getConsumer from the \"" << qname << "\" fail! No queue_consumer handle found!\n";
            return ConsumerPtr();
        }
        return que_consumer->getConsumer();
    }
    bool emptyQueue(const std::string &qname)   // 判断队列消费者是否为空
    {
        QueueConsumerPtr que_consumer = getQueueConsumerPtr(qname);
        if(que_consumer == nullptr) {
            LOG_ERROR << "unkonw if the \"" << qname << "\" is empty! No queue_consumer handle found!\n";
            return false;
        }
        return que_consumer->emptyQueue();
    }
    bool exists(const std::string& qname, const std::string& tag)  // 判断指定消费者是否存在
    {
        QueueConsumerPtr que_consumer = getQueueConsumerPtr(qname);
        if(que_consumer == nullptr) {
            LOG_ERROR << "unkonw if the " << qname << "'s conmuser is exists! No queue_consumer handle found!\n";
            return false;
        }
        return que_consumer->exists(tag);       
    }
    void clear()
    {
		std::lock_guard<std::mutex> lock(mtx_);
        que_consumers_.clear();
    }
private:
	QueueConsumerPtr getQueueConsumerPtr(const std::string& qname)
	{
		std::lock_guard<std::mutex> lock(mtx_);
		return existsQueue(qname) ? que_consumers_[qname] : nullptr;
	}
    bool existsQueue(const std::string& qname)
    {
        return que_consumers_.find(qname) != que_consumers_.end();
    }
private:
    std::mutex mtx_;
    std::unordered_map<std::string, QueueConsumerPtr> que_consumers_;  // <qname, QueueConsumer>
};

}

#endif